package cn.ai.boot.config;

import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.LongStream;

/**
 * @author jie.zhao@dms365.com
 * @Date 2025/6/26 14:27
 */
public class DaCbbQueryDataService {
    private static final Logger log = LoggerFactory.getLogger(DaCbbQueryDataService.class);
    private static long lastId = 0; // 初始 lastId
    private static final long pageSize = 100; // 每页大小
    private static final long totalItems = 1520; // 总共要模拟的数据量
    private static final int PAGE_SIZE = 100;

    public static List<Map<String, Object>> pageQueryDaCbbData(long lastId) {
//        if (lastId >= 122) {
//            return Collections.emptyList(); // 数据读取完毕
//        }
//        if (lastId > 100) {
//            throw new RuntimeException("122");
//        }
        System.out.println("开始查询委书记");
        // 模拟从 lastId + 1 开始生成 100 条数据
        return LongStream.rangeClosed(lastId + 1, Math.min(lastId + pageSize, totalItems))
                .mapToObj(id -> {
                    Map<String, Object> item = new HashMap<>();
                    item.put("MONGO_ID", id); // 模拟 MongoDB _id
                    item.put("name", "Item-" + id);
                    return item;
                })
                .collect(Collectors.toList());
    }

    private static AtomicLong totalCount = new AtomicLong(0);

    public static void main(String[] args) {

        // 4. 构造 transformer：打印或修改数据
        Consumer<List<Map<String, Object>>> transformer = dataList -> {
            System.out.println("Processing batch with size: " + dataList.size());
            for (Map<String, Object> item : dataList) {
                // 可以在这里对数据做转换处理
                item.put("processed", true);
            }
        };

        ExecutorService executorService = Executors.newFixedThreadPool(100);
        try {
            DeviceInitContext context = new DeviceInitContext();
            context.setUpdatedCompIds(Collections.synchronizedSet(new HashSet<>()));
            // 个性化异步
            queryTransformUpdateAsync(executorService, dataList -> handlePipeBasicAttr(context, dataList));
            // 通用异步
/*            queryTransformUpdateAsync555(dataList -> handlePipeBasicAttr(context, dataList),
                    executorService, "1111", 10);*/

/*            queryTransformUpdateAsync555(datas -> {
                List<Map<String, Object>> maps = new ArrayList<>();
                HashMap<String, Object> objectObjectHashMap = new HashMap<>();
                objectObjectHashMap.put("MONGO_ID", 999999999999999999L);
                maps.add(objectObjectHashMap);

                datas = maps;
            }, executorService, "1111", 10);*/

            // 通用删除
            /*queryTransformDeleteAsync(dataList -> {
                List<Long> ids = new ArrayList<>();
                for (Map<String, Object> x : dataList) {
                    ids.add(MapUtils.getLong(x, "MONGO_ID", 0L));
                    Long id = MapUtils.getLong(x, "MONGO_ID", 0L);
//                    if (id > 1000) {
//                        throw new RuntimeException("11");
//                    }
                }
                ids.addFirst(9999L);
                return ids;
            }, executorService, "deleteBeforeUpdateAll", 5);*/


            System.out.println("queryTransformUpdateAsync-out");
            executorService.shutdown();
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (Exception e) {
            System.out.println("11111111Exception");
        } finally {
            executorService.shutdown();
        }

        System.out.println("111111111111");


    }

    public static void handlePipeBasicAttr(DeviceInitContext context, List<Map<String, Object>> dataList) {
        Set<Long> updatedCompIds = context.getUpdatedCompIds();
        for (Map<String, Object> item : dataList) {
            updatedCompIds.add((Long) item.get("MONGO_ID"));
        }
        System.out.println("handlePipeBasicAttr" + updatedCompIds.size());
    }

    private static final Logger LOGGER = LoggerFactory.getLogger(DaCbbQueryDataService.class);

    public static void processInAsync(AsyncDataProcessor<?> processor,
                                      ExecutorService iThreadPoolService,
                                      String threadPoolName,
                                      int maxConcurrency) {
        Semaphore semaphore = new Semaphore(maxConcurrency);
        AtomicLong lastId = new AtomicLong(0L);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Future<Boolean>> futures = new ArrayList<>();
        try {
            while (true) {
                if (isException.get()) {
                    LOGGER.info("async.threadPoolName.{},run.break", threadPoolName);
                    throw new RuntimeException("processInAsync.isException.true");
                }
                LOGGER.debug("async.threadPoolName.{},run.start,availablePermits.size:{}", threadPoolName,
                        semaphore.availablePermits());
                semaphore.acquire();
                List<Map<String, Object>> dataList = pageQueryDaCbbData(
                        lastId.get()
                );
                if (org.apache.commons.collections4.CollectionUtils.isEmpty(dataList)) {
                    semaphore.release();
                    break;
                }
                LOGGER.debug("async.threadPoolName.{},run.start,GET.acquire:{}", threadPoolName, dataList.getLast());
                lastId.set(MapUtils.getLong(dataList.get(dataList.size() - 1), "MONGO_ID"));
                futures.add(iThreadPoolService.submit(() -> {
                    try {
                        I18nMessageUtils.initThreadLocale();
                        processor.process(dataList);
                        Locale locale1 = LocaleUtils.get();
                        System.out.println("initThreadLocale:" + locale1);
                    } catch (Exception e) {
                        LOGGER.error("processInAsync.submit.err", e);
                        isException.set(true);
                        throw e;
                    } finally {
                        semaphore.release();
                        I18nMessageUtils.remove();
                        Locale locale1 = LocaleUtils.get();
                        System.out.println("initThreadLocalefinally:" + locale1);
                    }
                    return true;
                }));
                if (dataList.size() < PAGE_SIZE) {
                    break;
                }
            }
            // 等待所有任务完成
            LOGGER.info("等待所有任务完成开始:{}..", threadPoolName);
            for (Future<Boolean> future : futures) {
                try {
                    future.get();
                } catch (Exception e) {
                    throw new RuntimeException("processInAsync.InterruptedException", e);
                }
            }
            LOGGER.info("等待所有任务完成结束:{}...", threadPoolName);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            isException.set(true);
            LOGGER.error("Processing interrupted", e);
        } catch (Exception e) {
            LOGGER.error("Processing interrupted", e);
            throw e;
        }
        if (isException.get()) {
            throw new RuntimeException("processInAsync.Exception");
        }
    }

    public static void queryTransformDeleteAsync(Function<List<Map<String, Object>>, List<Long>> transformer,
                                                 ExecutorService iThreadPoolService,
                                                 String threadPoolName,
                                                 int maxConcurrency) {
        processInAsync(dataList -> {
            List<Long> ids = transformer.apply(dataList);
            batchDeleteCbbData(ids);
        }, iThreadPoolService, threadPoolName, maxConcurrency);
    }

    public static void batchDeleteCbbData(List<Long> ids) {
        System.out.println("删除:" + ids + "成功");
        try {
            System.out.println("休眠一秒");
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 调用通用异常处理流程
     */
    public static void queryTransformUpdateAsync555(Consumer<List<Map<String, Object>>> transformer,
                                                    ExecutorService iThreadPoolService,
                                                    String threadPoolName,
                                                    int maxConcurrency) {
        processInAsync(dataList -> {
            transformer.accept(dataList);
            batchUpdateCbbData("DaCbbCodeConstant.EPC01", dataList);
        }, iThreadPoolService, threadPoolName, maxConcurrency);
    }

    /**
     * 可用
     */
    public static void queryTransformUpdateAsync(ExecutorService executorService,
                                                 Consumer<List<Map<String, Object>>> transformer) {

        final int MAX_CONCURRENT = 10;
        Semaphore semaphore = new Semaphore(MAX_CONCURRENT);
        AtomicLong lastId = new AtomicLong(0L);
        AtomicBoolean atomicBoolean = new AtomicBoolean(false);

        try {
            while (true) {

                semaphore.acquire(); // 先获取许可再查询数据
                if (atomicBoolean.get()) { // 如果发生异常则直接退出循环
                    break;
                }
                List<Map<String, Object>> dataList = pageQueryDaCbbData(
                        lastId.get()
                );

                if (CollectionUtils.isEmpty(dataList)) {
                    semaphore.release();
                    break;
                }

                lastId.set(MapUtils.getLong(dataList.get(dataList.size() - 1), "MONGO_ID"));

                executorService.submit(() -> {
                    try {
                        transformer.accept(dataList);
                        batchUpdateCbbData("DaCbbCodeConstant.EPC01", dataList);
                    } catch (Exception e) {
                        atomicBoolean.set(true);
                        System.out.println("queryTransformUpdateAsync.err" + e);
                    } finally {
                        semaphore.release();
                    }
                });


                if (dataList.size() < PAGE_SIZE) {
                    break;
                }
            }
            semaphore.acquire(MAX_CONCURRENT);
            System.out.println("所有任务执行完成");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            int i = semaphore.availablePermits();
            System.out.println("剩余可用许可证数量：" + i);
            semaphore.release(MAX_CONCURRENT);
        }

        if (atomicBoolean.get()) {
            System.out.println("线程异常");
            throw new RuntimeException("线程异常");
        }

    }


    public static void batchUpdateCbbData(String code, List<Map<String, Object>> dataList) {
        // 不执行真实更新，仅用于测试流程
        System.out.println("batchUpdateCbbData 被调用，数据条数：" + dataList.size());
        long l = totalCount.addAndGet(dataList.size());
        if (l >= 500) {
            System.out.println(1 / 0);
        }
        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
